﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PriceLibrary.PriceService;
using PriceLibrary.ChartService;
using PriceLibrary.Enums;
using PriceLibrary.Utilities;
using log4net;
using System.Threading;
using VDataAdapter.BlockingQueue;
using System.ServiceModel;
using System.Timers;

using VDataAdapter.Communication;
using VDataAdapter.ClientServer;

namespace VDataAdapter
{
    public delegate void ServiceConnected(bool isConnected);
    public class PSAdapterThread
    {
        #region Properties



        private static ILog _log = LogManager.GetLogger(typeof(PSAdapterThread));
        private readonly ManualResetEvent _terminalEvent;
        private EControlCodes HoseControlCode;
        
        private static PriceCallBack _PriceCallBack;
        public static PriceCallBack PriceCallBack
        {
            get { return _PriceCallBack; }
            set { _PriceCallBack = value; }
        }        
        private static ClientManager _ClientManager;
        public static ClientManager PriceManager
        {
            get { return _ClientManager; }
            set { _ClientManager = value; }
        }
        private static ChartCallbackManager _ChartCallbackManager;
        protected ChartCallbackManager TransactionManager
        {
            get { return _ChartCallbackManager; }
            set { _ChartCallbackManager = value; }
        }

        private static ChartCallBack _ChartCallBack;

        public ChartCallBack TransactionClient
        {
            get { return _ChartCallBack; }
            set { _ChartCallBack = value; }
        }
        protected VDataAdapter.Util.Setting ClientSetting = new VDataAdapter.Util.Setting();
        private DataSource _dataSource;
        private VDataClientThread vdataAdapter;
        // private MessageBuilder _msgBuilder = new MessageBuilder();
        private MessageBuilder _msgBuilder = new MessageBuilderSaver();

        public CommunicationDataQueue qMain;

        public CommunicationDataQueue _testQueue;

        private Queue<string>  queueTemp=new Queue<string>() ;
        //private List<string> tickers=new List<string>();
        private int _iCountTicker;

        private int iCountTransaction = 0;
        protected bool IsGetFirstTimeDone = false;
        private System.Timers.Timer timerService;

        private int intcountNumber = 0;
        private int intcountNumberTemp = 0;

        List<PriceLibrary.ChartService.Transaction> TransHolding = new List<PriceLibrary.ChartService.Transaction>();
        #endregion               

        #region Contructors
        public PSAdapterThread(ManualResetEvent resetEvent, CommunicationDataQueue qMain)
        {
            _terminalEvent = resetEvent;
            HoseControlCode = EControlCodes.K;
            this.qMain = qMain;
            _dataSource = new DataSource();
            
            // ThreadPool.QueueUserWorkItem(delegate { Init(); });
        }
        public void Start()
        {
            // Console.WriteLine("start: " + Thread.CurrentThread.ManagedThreadId);
            ThreadPool.QueueUserWorkItem(delegate { Init(); });
        }
        private void Init()
        {
            try
            {
                //Init client manager for all callback
                IsGetFirstTimeDone = false;
                _PriceCallBack = new PriceCallBack(ClientSetting.PriceURI);
                _ClientManager = new ClientManager(ClientSetting.PriceURI);
                _ChartCallbackManager = new ChartCallbackManager(ClientSetting.ChartURI);
                
                _ChartCallBack = new ChartCallBack(ClientSetting.ChartURI);

                
               //_ChartCallBack = _ChartCallbackManager.CreateChartCallbackWithServiceChannel();

                //Inits delegate for PriceClient                                
                _PriceCallBack.EndSendStockData += GetFirstTime;
                _PriceCallBack.Top3PriceChanged += Top3PricesStockChanged;
                _PriceCallBack.EndReceivedNewTickers += NewTickersReceived;
                _PriceCallBack.MarketInfoChanged += MarketInfo_Changed;
                _PriceCallBack.EndReceivedPtTopPrices += PtTopPrices_Changed;
                _PriceCallBack.EndReceivedPtTransactions += PtTransactions_Changed;
                //_PriceCallBack.ServiceStatusChanged += new ChangeServiceStatus(ChangeServiceStatus);

                //Gets tickers
                _PriceCallBack.PriceServer.GetTickersByFloorCode(100); // 100 get all
                // will receive a new ticker event: in order to get list of stocks
                _ChartCallBack.ChartService.SubscribeSymbols(new List<string>());  
                //Transaction service!
                _ChartCallBack.EndReceivedAllTransaction += ReceivedAllTransaction;
                
                //if (((ICommunicationObject)_ChartCallBack.ChartService).State == CommunicationState.Faulted)
                //{
                //    _ChartCallBack.ChartService = _ChartCallbackManager.CreateChartChannel(_ChartCallBack);
                //}
                PriceCallBack.IsInited = true;
                Reconnect();
            }
            catch (Exception ex)
            {
                _log.Error("Error Init", ex);
            }
        }        
        public void Reset()
        { 
            try
            {                                  
                _PriceCallBack.EndSendStockData -= GetFirstTime;
                _PriceCallBack.Top3PriceChanged -= Top3PricesStockChanged;
                _PriceCallBack.EndReceivedNewTickers -= NewTickersReceived;
                _PriceCallBack.MarketInfoChanged -= MarketInfo_Changed;
                _PriceCallBack.EndReceivedPtTopPrices -= PtTopPrices_Changed;
                _PriceCallBack.EndReceivedPtTransactions -= PtTransactions_Changed;

                _ChartCallBack.EndReceivedAllTransaction -= ReceivedAllTransaction;
                //_PriceCallBack.ServiceStatusChanged -= new ChangeServiceStatus(ChangeServiceStatus);
                Unsubcribes();
                RegisterEvents();
            }
            catch (Exception ex)
            {
                _log.Error("Error Init", ex);
            }
        }        
        
        #endregion Contructors

        #region Process Reconnect
        
        private void Reconnect()
        {
            try
            {
                timerService = new System.Timers.Timer();
                timerService.Interval = 60000;
                timerService.Elapsed += new ElapsedEventHandler((sender, e) =>
                {
                    try
                    {
                        ChartCallBackPing();
                        PriceCallBackPing();
                    }
                    catch
                    { }
                });
                timerService.Start();
            }
            catch
            {
                
            }
        }
        
        public static event ServiceConnected ServiceConnectedChanged;
        public static void InvokeServiceConnected(bool isConnected)
        {
            ServiceConnected handler = ServiceConnectedChanged;
            if (handler != null) handler(isConnected);
        }
        static void ChangeServiceStatus(bool isConnected)
        {
            InvokeServiceConnected(isConnected);
        }
        public static void PriceCallBackPing()
        {
            try
            {
                _log.ErrorFormat("Ping Price Server at {0}", DateTime.Now.ToString());  
                if (((ICommunicationObject)_PriceCallBack.PriceServer).State == CommunicationState.Faulted)
                {
                    Thread.Sleep(500);
                    _log.ErrorFormat("CommunicationState.Faulted State at {0}", DateTime.Now.ToString());  
                    _PriceCallBack.PriceServer = _ClientManager.CreatePriceServer(_PriceCallBack);
                } 
                 _PriceCallBack.PriceServer.SubscribeNewSymbols(new List<string>());
            }
            catch(Exception ex)
            {
                _log.Error("Error PriceCallBackPing", ex);
            }
        }
        public static void ChartCallBackPing()
        {
            try
            {
                if (((ICommunicationObject)_ChartCallBack.ChartService).State == CommunicationState.Faulted)
                {
                    Thread.Sleep(500);
                    _ChartCallBack.ChartService = _ChartCallbackManager.CreateChartChannel(_ChartCallBack);
                }
                //else Console.WriteLine(">>Inited Chart Services");
                _ChartCallBack.ChartService.SubscribeSymbols(new List<string>());
            }
            catch
            {
                Console.WriteLine(">>Init Chart Services Fail");
            }
        }
        #endregion




        #region Event Handler
        /// <summary>
        /// Get Stock list when fisrt running 
        /// </summary>
        /// <param name="baseprices"></param>
        /// <param name="top3Prices"></param>
        protected virtual void GetFirstTime(IDictionary<string, PriceLibrary.PriceService.StockBase> baseprices,IDictionary<string, PriceLibrary.PriceService.StockBase> top3Prices)
        {
            
            foreach (KeyValuePair<string, PriceLibrary.PriceService.StockBase> p in baseprices)
            {
                var infoStatic =
                    (PriceLibrary.PriceService.StockInfoStatic)p.Value;
                _dataSource.UpdateStockBase(infoStatic);

                
                
            }
            foreach (KeyValuePair<string, PriceLibrary.PriceService.StockBase> p in top3Prices)
            {
                var infoDynamic =
                    (PriceLibrary.PriceService.StockInfoDyns)p.Value;
                _dataSource.UpdateStockDynamic(infoDynamic);


            }
            _iCountTicker = baseprices.Count;
            if (_iCountTicker > 0)
            {
                //BEGIN
                ProcessMessageDay("BEGIN"); 
                //ProcessStockInfo
                ProcessStockInfo(top3Prices);
                //END
                ProcessMessageDay("END");

                IsGetFirstTimeDone = true;

                MessageBuilderRecorder recorder = new MessageBuilderRecorder();
                recorder.WriteDataSourceStockbase(_dataSource);

                while (queueTemp.Count > 0)
                {
                    var data = queueTemp.Dequeue();
                    UpdateQueue(data);
                }                
            }
        }        
        
        /// <summary>
        /// Received New Tickers and SubscribeNewSymbols to PriceClient.PriceServer and PriceClient.ChartService
        /// </summary>
        /// <param name="newtickers"></param>
        private void NewTickersReceived(IList<string> newtickers)
        {
            try
            {
                if (_PriceCallBack != null)
                {
   
                    foreach (string newticker in newtickers)
                    {
                        _PriceCallBack.Tickers.Add(newticker);
                        TransactionClient.Symbols.Add(newticker);
          
                        //tickers.Add(newticker);
                    }

                    //Đăng ký lên server!
                    _PriceCallBack.PriceServer.SubscribeNewSymbols(newtickers as List<string>);

                    newtickers = AddMarketTickers(newtickers);

                    //_ChartCallBack.ChartService.d  
                    _ChartCallBack.ChartService.SubscribeAllSymbols((List<string>)newtickers);
                }
            }
            catch(Exception ex)
            {
                _log.Error("Error NewTickersReceived ", ex);
            }
        }

        /// <summary>
        /// Add symbols of market to newtickers
        /// </summary>
        /// <param name="newtickers"></param>
        /// <returns></returns>
        private static IList<string> AddMarketTickers(IList<string> newtickers)
        {
            if (!newtickers.Contains("HOSE"))
                newtickers.Add("HOSE");
            if (!newtickers.Contains("HNX"))
                newtickers.Add("HNX");
            if (!newtickers.Contains("UPCOM"))
                newtickers.Add("UPCOM");
            if (!newtickers.Contains("VN30"))
                newtickers.Add("VN30");
            if (!newtickers.Contains("HNX30"))
                newtickers.Add("HNX30");

            return newtickers;
        }
        /// <summary>
        /// Received Securities Info Changed event
        /// </summary>
        /// <param name="top3Prices"></param>
        protected virtual void Top3PricesStockChanged(IDictionary<string, PriceLibrary.PriceService.StockBase> top3Prices)
        {
            try
            {
                if (IsGetFirstTimeDone)
                {
                    _log.Info("SecInfo Changed!");
                    var thread = new Thread(ProcessStockInfo);
                    thread.Start(top3Prices);
                    //ProcessStockInfo(top3Prices);
                }
            }
            catch (Exception ex)
            {
                _log.Error(ex);
            }

        }
        /// <summary>
        /// Received MarketInfo Changed event
        /// </summary>
        /// <param name="maketinfos"></param>
        protected virtual void MarketInfo_Changed(IDictionary<int, PriceLibrary.PriceService.StockBase> maketinfos)
        {
            try
            {
                //if (IsGetFirstTimeDone)
                {
                    _log.Info("MarketInfo_Changed!");
                    var thread = new Thread(ProcessMarketInfo);
                    thread.Start(maketinfos);
                    //ProcessMarketInfo(maketinfos);
                }
            }
            catch (Exception ex)
            {
                _log.Error(ex);
            }
        }
        /// <summary>
        /// Received Transaction Changed event
        /// </summary>
        /// <param name="transactions"></param>
        protected virtual void ReceivedAllTransaction(IList<PriceLibrary.ChartService.Transaction> transactions, int countNumber, int totalCountNumber)
        {
            try
            {
                //if (IsGetFirstTimeDone)
                {
                    //intcountNumber = countNumber;
                    //intcountNumberTemp = intcountNumberTemp + 1;
                    _log.Info("ReceivedAllTransaction!");
                    //var thread = new Thread(ProcessTransactionInfo);
                    //thread.Start(transactions);
                    ProcessTransactionInfo(transactions, countNumber, totalCountNumber);
                }
            }
            catch (Exception ex)
            {
                _log.Error(ex);
                throw new Exception("Error While Processing Transaction Info");
            }
        }
        /// <summary>
        /// Received PutThrough Transaction Changed event
        /// </summary>
        /// <param name="newstockbase"></param>
        protected virtual void PtTransactions_Changed(IList<PriceLibrary.PriceService.StockBase> newstockbase)
        {
            try
            {
                //if (IsGetFirstTimeDone)
                {
                    _log.Info("PtTransactions Changed!");
                    var thread = new Thread(ProcessPtTransaction);
                    thread.Start(newstockbase);
                }
            }
            catch (Exception ex)
            {
                _log.Error(ex);
            }
        }
        /// <summary>
        /// Received PutThrough Stock Changed event
        /// </summary>
        /// <param name="newstockbase"></param>
        protected virtual void PtTopPrices_Changed(IList<PriceLibrary.PriceService.StockBase> newstockbase)
        {
            try
            {
                //if (IsGetFirstTimeDone)
                {
                    _log.Info("PtTopPrice Changed!");
                    var thread = new Thread(ProcessPtTopPrices);
                    thread.Start(newstockbase);
                    //ProcessPtTopPrices(newstockbase);
                }                
            }
            catch (Exception ex)
            {
                _log.Error(ex);
            }
        }

        #endregion Event Handler

        #region Functions



        /// <summary>
        /// ProcessStockInfo include BuildMessageStock, Add Msg to Queue and UpdateStock DataSource
        /// </summary>
        /// <param name="obj"></param>
        private void ProcessStockInfo(object obj)
        {
            int count = 0;
            try
            {

                var top3Prices = (IDictionary<string, PriceLibrary.PriceService.StockBase>)obj;
                foreach (KeyValuePair<string, PriceLibrary.PriceService.StockBase> p in top3Prices)
                {
                    var infoDynamic =
                        (PriceLibrary.PriceService.StockInfoDyns)p.Value;
                    //object infoStatic;
                    // if (dataSource.BasePrices.ContainsKey(p.Key) )
                    var infoStatic = (PriceLibrary.PriceService.StockInfoStatic)_dataSource.BasePrices[p.Key];
                    var infoDynamicOld =
                       (PriceLibrary.PriceService.StockInfoDyns)_dataSource.Top3Prices[p.Key];

                    var strMsg = _msgBuilder.BuildMessageStock(infoStatic, infoDynamic, infoDynamicOld);


                    //if (infoStatic.StockSymbol == "FPT" || infoStatic.StockSymbol == "SSI"
                    //    || infoStatic.StockSymbol == "HAG" || infoStatic.StockSymbol == "STB"
                    //    || infoStatic.StockSymbol == "ITA" || infoStatic.StockSymbol == "CTG" 
                    //    || infoStatic.StockSymbol == "REE"  )
                    count++;
                    UpdateQueue(strMsg);
                    _dataSource.UpdateStockDynamic(infoDynamic);
                }
                Console.WriteLine(count.ToString());
                count = 0;
            }
            catch (KeyNotFoundException ex)
            {
                throw new Exception("PSAdapter key not found: " + ex.Data.Keys.ToString());
            }
            catch (Exception ex)
            {
                throw new Exception("Unknown exception processing stock first time: " + ex.Message);
            }
            //_log.WarnFormat("Send Msg {0}", strMsg);

        }
        
        private string BuildMarketInfoMessage(MessageBuilder builder, int key, PriceLibrary.PriceService.MarketInfo newMarketInfo, DataSource dataSource)
        { 
            if (dataSource.MarketInfos.ContainsKey(key))
            {
                PriceLibrary.PriceService.MarketInfo oldMarketInfo = (PriceLibrary.PriceService.MarketInfo)dataSource.MarketInfos[key];
                return builder.BuildMessageMarket(oldMarketInfo, newMarketInfo);
            }
            else
            {
                return builder.BuildMessageMarket(newMarketInfo, newMarketInfo);
            }
        }

        private List<string> BuildMarketInfoMessages(IDictionary<int, PriceLibrary.PriceService.StockBase> marketInfos)
        {
            string strMsg;

            List<string> marketInfoList = new List<string>();

            foreach (KeyValuePair<int, PriceLibrary.PriceService.StockBase> p in marketInfos)
            {    
                PriceLibrary.PriceService.MarketInfo marketInfoNew = (PriceLibrary.PriceService.MarketInfo)p.Value;
                marketInfoNew.TotalShares *= 10;
                strMsg = BuildMarketInfoMessage(_msgBuilder, p.Key, marketInfoNew, _dataSource);
                marketInfoList.Add(strMsg);

                if (marketInfoNew != null)
                {
                    _dataSource.UpdateMarketInfo(marketInfoNew);
                }
            }
            return marketInfoList;
        }


        protected virtual void ProcessMarketInfo(IDictionary<int, PriceLibrary.PriceService.StockBase> marketInfos)
        {
            List<string> list = BuildMarketInfoMessages(marketInfos);

            //while (!IsGetFirstTimeDone)
            //{
            //    Thread.Sleep(100);
            //}

            foreach (string msg in list)
            {
                UpdateQueue(msg);
            }
        }
        
        protected virtual void ProcessMarketInfoOld(IDictionary<int, PriceLibrary.PriceService.StockBase> marketInfos)
        {
            try
            {
                foreach (KeyValuePair<int, PriceLibrary.PriceService.StockBase> p in marketInfos)
                {
                    //if (_terminalEvent.WaitOne(0, false)) return;
                    int key = p.Key;
                    PriceLibrary.PriceService.MarketInfo marketInfoNew = (PriceLibrary.PriceService.MarketInfo)p.Value;
                    marketInfoNew.TotalShares *= 10;

                    string strMsg = BuildMarketInfoMessage(_msgBuilder, key, marketInfoNew, _dataSource);

                    //var marketInfoNew = (PriceLibrary.PriceService.MarketInfo)p.Value;

                    //var marketInfoOld = marketInfoNew;

                    //if (_dataSource.MarketInfos.ContainsKey(p.Key))
                    //    marketInfoOld = (PriceLibrary.PriceService.MarketInfo)_dataSource.MarketInfos[p.Key];

                    //var strMsg = msgbuider.BuildMessageMarket(marketInfoNew, marketInfoOld);

                    //UpdateQueue(strMsg);
                    //dataSource.UpdateMarketInfo(marketInfoNew);
                    //WriteMsgToFile(strMsg);

                    if (IsGetFirstTimeDone && marketInfoNew != null)
                    {
                        UpdateQueue(strMsg);
                        _dataSource.UpdateMarketInfo(marketInfoNew);
                        //WriteMsgToFile(strMsg);
                    }
                    else if (marketInfoNew != null)
                    {
                        UpdateQueueTemp(strMsg);
                        _dataSource.UpdateMarketInfo(marketInfoNew);
                    }



                }
            }
            catch (Exception ex)
            {
                _log.Error("Error ProcessMarketInfo:", ex);
            }
        }

        /// <summary>
        /// ProcessMarketInfo include BuildMessageMarket, Add Msg to Queue and UpdateMarketInfo DataSource
        /// </summary>
        /// <param name="obj"></param>
        protected virtual void ProcessMarketInfo(object obj)
        {
            try
            {
                var maketinfos = (IDictionary<int, PriceLibrary.PriceService.StockBase>)obj;

                ProcessMarketInfo(maketinfos);

                //foreach (KeyValuePair<int, PriceLibrary.PriceService.StockBase> p in maketinfos)
                //{
                //    //if (_terminalEvent.WaitOne(0, false)) return;
                //    int key = p.Key;
                //    PriceLibrary.PriceService.MarketInfo marketInfoNew = (PriceLibrary.PriceService.MarketInfo)p.Value;
                //    marketInfoNew.TotalShares *= 10;

                //    string strMsg = BuildMarketInfoMessage(_msgBuilder, key, marketInfoNew, _dataSource);

                //    //var marketInfoNew = (PriceLibrary.PriceService.MarketInfo)p.Value;
                    
                //    //var marketInfoOld = marketInfoNew;

                //    //if (_dataSource.MarketInfos.ContainsKey(p.Key))
                //    //    marketInfoOld = (PriceLibrary.PriceService.MarketInfo)_dataSource.MarketInfos[p.Key];

                //    //var strMsg = msgbuider.BuildMessageMarket(marketInfoNew, marketInfoOld);
                    
                //    //UpdateQueue(strMsg);
                //    //dataSource.UpdateMarketInfo(marketInfoNew);
                //    //WriteMsgToFile(strMsg);

                //    if (IsGetFirstTimeDone && marketInfoNew != null)
                //    {
                //        UpdateQueue(strMsg);
                //        _dataSource.UpdateMarketInfo(marketInfoNew);
                //        //WriteMsgToFile(strMsg);
                //    }
                //    else if (marketInfoNew != null)
                //    {
                //        UpdateQueueTemp(strMsg);
                //        _dataSource.UpdateMarketInfo(marketInfoNew);
                //    }
                //}
            }
            catch (Exception ex)
            {
                _log.Error("Error ProcessMarketInfo:", ex);
            }
        }

        /// <summary>
        /// Lay tat ca msg trong queue va update vao queueMain de send
        /// </summary>
        /// <param name="queue"></param>
        private void SendAllMsgInQueue(Queue<string> queue)
        {
            while (queue.Count > 0)
            {
                var data = queue.Dequeue();
                UpdateQueue(data);
            }       
        }

        private Queue<string> BuildTransactionsQueue(DataSource dataSource, List<PriceLibrary.ChartService.Transaction> transactions)
        {
            Queue<string> transactionsQueue = new Queue<string>();
           string strMsg = "";

            foreach (PriceLibrary.ChartService.Transaction transaction in transactions)
            {
                var transactionOld = transaction;
                
                if (dataSource.Transactions.ContainsKey(transaction.Symbol))
                    transactionOld = dataSource.Transactions[transaction.Symbol];

                strMsg = _msgBuilder.BuildMessageTransaction(transaction, transactionOld);
                transactionsQueue.Enqueue(strMsg);
            }

            return transactionsQueue;
        }

        /// <summary>
        /// ProcessTransactionInfo include BuildMessageTransactionInfo, Add Msg to Queue and UpdateTransactionInfo DataSource
        /// </summary>
        /// <param name="transactionsList"></param>
        protected virtual void ProcessTransactionInfo(IList<PriceLibrary.ChartService.Transaction> transactionsList, int countNumber, int totalCountNumber)
        {
            // add transactionsLilst to TransHolding
            TransHolding = BuildTransactionsHolding(transactionsList, TransHolding);

            if (countNumber == totalCountNumber)
            {
                // order transations by time

                List<PriceLibrary.ChartService.Transaction>  collection = TransHolding.Distinct().OrderBy(c => int.Parse(c.Time)).ToList();
                //List<PriceLibrary.ChartService.Transaction> transaction=(List<PriceLibrary.ChartService.Transaction>) collection;
                // build a queue of string from list of transactions
                Queue<string> transactionsQueue = BuildTransactionsQueue(_dataSource, collection);
                // wait for get first time done
                while (!IsGetFirstTimeDone) ;
                // send all items in queue
                SendAllMsgInQueue(transactionsQueue);
                // update END message for transactions
                string strMsg = "END_RECOVERY" + "#_#_";
                UpdateQueue(strMsg);
                TransHolding.Clear();
                Console.WriteLine("End received Transactions");
                
                //foreach (PriceLibrary.ChartService.Transaction transaction in collection)
                //{
                //    var transactionOld = transaction;
                    
                //    if (_dataSource.Transactions.ContainsKey(transaction.Symbol))
                //        transactionOld = _dataSource.Transactions[transaction.Symbol];

                //    strMsg = msgbuider.BuildMessageTransaction(transaction, transactionOld);r

                //    while (!IsGetFirstTimeDone) ; // wait for get first time done
                                         

                //    if (IsGetFirstTimeDone)
                //    {
                //        UpdateQueue(strMsg);
                //        _dataSource.UpdateTransaction(transaction);
                //    }
                //    else
                //    {
                //        UpdateQueueTemp(strMsg);
                //        _dataSource.UpdateTransaction(transaction);
                //    }
                //}
            }
            //if (IsGetFirstTimeDone && countNumber == totalCountNumber)  
            //{
            //    strMsg = "END_RECOVERY" + "#_#_";
            //    UpdateQueue(strMsg);
            //}
            //else if (!IsGetFirstTimeDone && countNumber == totalCountNumber) 
            //{
            //    strMsg = "END_RECOVERY" + "#_#_";
            //    UpdateQueueTemp(strMsg);
            //}
        }

        private List<PriceLibrary.ChartService.Transaction> BuildTransactionsHolding(IList<PriceLibrary.ChartService.Transaction> transactions, List<PriceLibrary.ChartService.Transaction> transHolding)
        {
            // var transactions = (IList<PriceLibrary.ChartService.Transaction>)obj;
            int iCountTransaction = transactions.Count;
            var strMsg = "";
            foreach (PriceLibrary.ChartService.Transaction transaction in transactions)
            {
                //if (_terminalEvent.WaitOne(0, false)) return;
                //if (transaction.Symbol != "HOSE"
                // && transaction.Symbol != "HNX"
                // && transaction.Symbol != "UPCOM")
                //if (transaction.Symbol == "UPCOM")
                {
                    iCountTransaction = iCountTransaction - 1;
                    //if (transaction.Symbol=="KLS")
                    transHolding.Add(transaction);
                }
            }
            return transHolding;
        }

        protected virtual void ProcessPtTopPrices(IList<PriceLibrary.PriceService.StockBase> newstockbase)
        {

            foreach (PriceLibrary.PriceService.StockBase stockBase in newstockbase)
            {
                var infoPtPrice =
                    (PriceLibrary.PriceService.StockInfoDyns)stockBase;

                if (_dataSource.BasePrices.ContainsKey(infoPtPrice.StockSymbol))
                {
                    var infoStatic =
                      (PriceLibrary.PriceService.StockInfoStatic)_dataSource.BasePrices[infoPtPrice.StockSymbol];

                    if (infoPtPrice.Best1BidVolume > 0)
                    {
                        var infoPtPriceBuyOld = infoPtPrice;
                        if (_dataSource.PtTopPriceBuy.ContainsKey(infoPtPrice.StockSymbol))
                            infoPtPriceBuyOld = (PriceLibrary.PriceService.StockInfoDyns)_dataSource.PtTopPriceBuy[infoPtPrice.StockSymbol];
                        
                        var strMsg = _msgBuilder.BuildMessageBuyPutThrough(infoStatic, infoPtPrice, infoPtPriceBuyOld);
                        //UpdateQueue(strMsg);
                        //dataSource.UpdatePutTopPriceBuy(infoPtPrice);
                        if (IsGetFirstTimeDone)
                        {
                            UpdateQueue(strMsg);
                            _dataSource.UpdatePutTopPriceBuy(infoPtPrice);
                            WriteMsgToFile(strMsg);
                        }
                        else
                        {
                            UpdateQueueTemp(strMsg);
                            _dataSource.UpdatePutTopPriceBuy(infoPtPrice);
                        }
                    }

                    //Check Best1OfferVolume>0
                    if (infoPtPrice.Best1OfferVolume > 0)
                    {
                        var infoPtPriceSellOld = infoPtPrice;
                        if (_dataSource.PtTopPriceSell.ContainsKey(infoPtPrice.StockSymbol))
                            infoPtPriceSellOld = (PriceLibrary.PriceService.StockInfoDyns)_dataSource.PtTopPriceSell[infoPtPrice.StockSymbol];

                        var strMsg = _msgBuilder.BuildMessageSellPutThrough(infoStatic, infoPtPrice, infoPtPriceSellOld);
                        //UpdateQueue(strMsg);
                        //dataSource.UpdatePutTopPriceSell(infoPtPrice);
                        if (IsGetFirstTimeDone)
                        {
                            UpdateQueue(strMsg);
                            _dataSource.UpdatePutTopPriceSell(infoPtPrice);
                            WriteMsgToFile(strMsg);
                        }
                        else
                        {
                            UpdateQueueTemp(strMsg);
                            _dataSource.UpdatePutTopPriceSell(infoPtPrice);
                        }
                    }
                }


            }
        }

        /// <summary>
        /// ProcessPtTopPrices include
        /// If Best1BidVolume > 0 then 
        ///     BuildMessageBuyPutThrough --- Add Msg to Queue and UpdateBuyPutThrough DataSource
        /// If Best1OfferVolume > 0 then 
        ///     BuildMessageSellPutThrough -- Add Msg to Queue and UpdateSellPutThrough DataSource        
        /// </summary>
        /// <param name="obj"></param>
        private void ProcessPtTopPrices(object obj)
        {
            var newstockbase = (IList<PriceLibrary.PriceService.StockBase>)obj;

            ProcessPtTopPrices(newstockbase);
        }

        

        /// <summary>
        /// ProcessPtTransaction include BuildMessageTransactionPutThrough,  Add Msg to Queue and UpdatePutThroughTransaction DataSource  
        /// </summary>
        /// <param name="obj"></param>
        private void ProcessPtTransaction(object obj)
        {
            var transactions = (IList<PriceLibrary.PriceService.StockBase>)obj;

            ProcessPtTransaction(transactions);
        }

        protected virtual void ProcessPtTransaction(IList<PriceLibrary.PriceService.StockBase> transactions)
        {
            foreach (PriceLibrary.PriceService.StockBase stockbase in transactions)
            {
                //if (_terminalEvent.WaitOne(0, false)) return;

                var transaction = (PriceLibrary.PriceService.Transaction)stockbase;

                var transactionOld = transaction;
                if (_dataSource.PtTransactions.ContainsKey(transaction.Symbol))
                    transactionOld = _dataSource.PtTransactions[transaction.Symbol];

                if (transactionOld == null)
                    transactionOld = transaction;
                var strMsg = _msgBuilder.BuildMessageTransactionPutThrough(transaction, transactionOld);
                //UpdateQueue(strMsg);
                //dataSource.UpdatePtTransaction(transaction);
                if (IsGetFirstTimeDone)
                {
                    UpdateQueue(strMsg);
                    _dataSource.UpdatePtTransaction(transaction);
                    WriteMsgToFile(strMsg);
                }
                else
                {
                    UpdateQueueTemp(strMsg);
                    _dataSource.UpdatePtTransaction(transaction);
                }
            }
        }

        /// <summary>
        /// To UPPER CASE and
        /// Add "#_#_" to the end of the message
        /// </summary>
        /// <param name="msg"></param>
        /// <returns></returns>
        private string BuildSignalMessage(string msg)
        {
            return msg.ToUpper() + "#_#_";
        }

        private void ProcessMessageDay(string strMessage)
        {
            var strMsg = strMessage.ToUpper() + "#_#_";
            UpdateQueue(strMsg);
            WriteMsgToFile(strMsg);
        }

        private void ProcessMessageEndRecovery(string strMessage)
        {
            var strMsg = strMessage.ToUpper() + "#_#_";
            UpdateQueue(strMsg);
            WriteMsgToFile(strMsg);
        }
        
        protected Queue<string> _strQueue = new Queue<string>();
        
        public Queue<string> StrQueue
        {
            get { return _strQueue; }
        }

        public void ClearStrQueue() { _strQueue.Clear(); }

        /// <summary>
        /// Add Msg to Queue
        /// </summary>
        /// <param name="strMsg"></param>
        private void UpdateQueue(string strMsg)
        {
            MessageObject objData = new MessageObject(strMsg);
            DataQueue objQueue = new DataQueue();
            this.qMain.put(new CommunicationData(objData, objQueue));
            _strQueue.Enqueue(strMsg);
            // this._testQueue.put(new CommunicationData(objData, objQueue));
            // WriteMsgToFile(strMsg);
        }

        private void UpdateQueueTemp(string strMsg)
        {
            queueTemp.Enqueue(strMsg);
        }
        
        private void RegisterEvents()
        {
            _PriceCallBack.EndSendStockData += GetFirstTime;
            _PriceCallBack.Top3PriceChanged += Top3PricesStockChanged;
            _PriceCallBack.EndReceivedNewTickers += NewTickersReceived;
            _PriceCallBack.MarketInfoChanged += MarketInfo_Changed;
            _PriceCallBack.EndReceivedPtTopPrices += PtTopPrices_Changed;
            _PriceCallBack.EndReceivedPtTransactions += PtTransactions_Changed;

            //Gets tickers
            //_PriceCallBack.PriceServer.GetTickersByFloorCode(100);
            _PriceCallBack.PriceServer.GetMarketInfos(true);
            //Transaction service!                
            TransactionClient.EndReceivedAllTransaction += ReceivedAllTransaction;
            _PriceCallBack.ServiceStatusChanged += new ChangeServiceStatus(ChangeServiceStatus);
        }
        #endregion

        #region Unsubcribes

        /// <summary>
        /// Unsubcribeses all stock symbols.
        /// </summary>
        public void Unsubcribes()
        {
            try
            {
                _PriceCallBack.PriceServer.UnsubscribeSymbols((List<string>)_PriceCallBack.Tickers);
                _PriceCallBack.PriceServer.GetMarketInfos(false);
                TransactionClient.ChartService.UnsubscribeAll();
            }
            catch (Exception ex)
            {
                _log.ErrorFormat("Error Unscubcribes", ex);
            }
        }

        #endregion Unsubcribes

        private void WriteMsgToFile(string sMessage)
        {
            //_log.WarnFormat(sMessage);
        }
    }
}
